package com.zzw.java.mqtt;


import cn.hutool.core.lang.TypeReference;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.zzw.java.pojo.MqttResponseVo;
import com.zzw.java.serial.ReceiveDataHandler;
import com.zzw.java.serial.ReceiveDataHandlerImpl;
import lombok.extern.slf4j.Slf4j;
import net.sf.jsqlparser.expression.JsonAggregateFunction;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;

/**
 * mqtt入站通道
 *
 * @Author：hcy
 * @Package：com.zzw.java.mqtt
 * @version：1.0
 * @Date：2025/8/8 17:37
 */


@Configuration
@Slf4j
public class MqttInBoundConfiguration {
    @Resource
    private MqttConfiguration mqttProperties;
    @Resource
    private ReceiveDataHandler receiveDataHandler;

    public static String sendTopic = "/dtu/${deviceId}/message/event/uploadData";

    /**
     * mis响应数据topic
     */
    public static String misResponseTopic = "/responseMisData";

    /**
     * 初始设置
     */
    @PostConstruct
    private void init() {
        // 获取Linux CPU序列号最为设备ID(替换以前的MAC码)
        try {
            // 执行命令
                Process process = Runtime.getRuntime().exec(
                    new String[] {"/bin/sh", "-c", "cat /proc/cpuinfo | grep Serial | awk '{print $3}'"}
            );

            process.getOutputStream().close();

            // 读取命令输出
            BufferedReader reader = new BufferedReader(
                    new InputStreamReader(process.getInputStream())
            );

            String deviceId = reader.readLine();
//             String deviceId = "181b29d4ddcc669c";
            if (deviceId != null) {
                log.info("CPU序列号: " + deviceId);
                sendTopic = sendTopic.replace("${deviceId}", deviceId);
                mqttProperties.getReceiveTopic()[0] = mqttProperties.getReceiveTopic()[0].replace("${deviceId}", deviceId);
                mqttProperties.getReceiveTopic()[1] = mqttProperties.getReceiveTopic()[1].replace("${deviceId}", deviceId);
                mqttProperties.setWillTopic(mqttProperties.getWillTopic().replace("${deviceId}", deviceId));
                ReceiveDataHandlerImpl.TRANSFER_MODULE_MAC = deviceId;
            } else {
                log.error("无法获取CPU序列号");
            }

        } catch (IOException e) {
            log.error("获取CPU序列号异常", e);
        }
    }


    //==================================== 消费消息==========================================//

    /**
     * 入站通道
     *
     * @return 消息通道对象 {@link MessageChannel}
     */
    @Bean("input")
    public MessageChannel mqttInputChannel() {
        // 直连通道
        return new DirectChannel();
    }


    /**
     * 创建MqttPahoClientFactory 设置MQTT的broker的连接属性 如果使用ssl验证 也需要此处设置
     *
     * @return MQTT客户端工厂 {@link MqttPahoClientFactory}
     */
    @Bean
    public MqttPahoClientFactory inClientFactory() {
        // 设置连接属性
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(mqttProperties.getHostUrl());
        options.setUserName(mqttProperties.getUsername());
        options.setPassword(mqttProperties.getPassword().toCharArray());
        options.setConnectionTimeout(mqttProperties.getTimeout());
        options.setKeepAliveInterval(mqttProperties.getKeepalive());
        // 接受离线消息  告诉代理客户端是否要建立持久会话   false为建立持久会话
        options.setCleanSession(false);
        // 设置断开后重新连接
        options.setAutomaticReconnect(true);
        String willPayload = "{\"isDisconnected\":1}";
        options.setWill(mqttProperties.getWillTopic(), willPayload.getBytes(StandardCharsets.UTF_8), 1, false);
        factory.setConnectionOptions(options);
        return factory;
    }


    /**
     * 入站
     *
     * @return 消息提供者 {@link MessageProducer}
     */
    @Bean
    public MessageProducer producer() {
        // Paho客户端消息驱动通道适配器，主要用来订阅主题  对inboundTopics主题进行监听
        // clientId 加后缀 不然会报retrying 不能重复
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(mqttProperties.getClientId() + "_customer", inClientFactory(), mqttProperties.getReceiveTopic());
        adapter.setCompletionTimeout(5000);
        // Paho消息转换器
        DefaultPahoMessageConverter defaultPahoMessageConverter = new DefaultPahoMessageConverter();
        // 按字节接收消息
        // defaultPahoMessageConverter.setPayloadAsBytes(true);
        adapter.setConverter(defaultPahoMessageConverter);
        // 设置QoS
        adapter.setQos(mqttProperties.getQos());
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    /**
     * 通过通道获取数据
     * ServiceActivator注解表明：当前方法用于处理MQTT消息，inputChannel参数指定了用于消费消息的channel。
     * tips:
     * 异步处理
     *
     * @return 消息处理 {@link MessageHandler}
     */
    @Bean
    @ServiceActivator(inputChannel = "input")
    public MessageHandler handler() {
        return message -> {
            // log.info("收到的完整消息为--->{}", message);
            log.info("----------------------");
            log.info("message:" + message.getPayload());
            // log.info("Id:" + message.getHeaders().getId());
            // log.info("receivedQos:" + message.getHeaders().get(MqttHeaders.RECEIVED_QOS));
            String topic = (String) message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC);
            log.info("topic:" + topic);
            log.info("----------------------");
            if (StrUtil.isNotEmpty(topic)&&ObjectUtil.isNotEmpty(message)) {
                Object payload = message.getPayload();
                if (ObjectUtil.isNotEmpty(payload)) {
                    // 使用 TypeReference 明确指定泛型类型
                    MqttResponseVo vo = JSONUtil.toBean(
                            payload.toString(),
                            new TypeReference<MqttResponseVo>() {},
                            true
                    );
                    if (ObjectUtil.isNotEmpty(vo)) {
                        receiveDataHandler.handlerMqttHistoricalDataData(vo.getData());
                    }
                }
            }
               // receiveDataHandler.handlerMqttData(message.getPayload().toString());
        };
    }
}


